package com.yy.distributed.lock.redis;

import com.yy.distributed.lock.DLock;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

/**
 * Jedis锁实现类
 *
 * @author dinghuan
 * @date 2018/7/19
 */
public class JedisLock implements DLock {
    public static final long LOCK_EXPIRATION_INTERVAL_SECONDS = 30;

    private final JedisPool jedisPool;
    private String lockName;

    protected long innerLeaseTime = TimeUnit.SECONDS.toMillis(LOCK_EXPIRATION_INTERVAL_SECONDS);

    protected boolean isLocked = false;
    protected final UUID id = UUID.randomUUID();
    protected final Semaphore semaphore = new Semaphore(1);

    public JedisLock(JedisPool jedisPool, String lockName) {
        this.jedisPool = jedisPool;
        this.lockName = lockName;
    }

    @Override
    public void lock() {
        try {
            lockInterruptibly();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    public void lock(long leaseTime, TimeUnit unit) {
        try {
            lockInterruptibly(leaseTime, unit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        lockInterruptibly(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS);
    }

    @Override
    public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
        Long ttl = tryAcquire(leaseTime, unit, Thread.currentThread().getId());
        if (ttl == null) {
            isLocked = true;
            return;
        }

        try {
            while (true) {
                ttl = tryAcquire(leaseTime, unit, Thread.currentThread().getId());
                if (ttl == null) {
                    isLocked = true;
                    break;
                }

                if (ttl >= 0) {
                    semaphore.tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    semaphore.acquire();
                }
            }
        } finally {
            if (semaphore.availablePermits() == 0) {
                semaphore.release();
            }
        }
    }

    @Override
    public boolean tryLock() {
        Long ttl = tryAcquire(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS, Thread.currentThread().getId());

        if (ttl == null) {
            isLocked = true;
            return true;
        }
        return false;
    }

    @Override
    public boolean tryLock(long waitTime, TimeUnit unit) {
        return tryLock(waitTime, LOCK_EXPIRATION_INTERVAL_SECONDS, unit);
    }


    @Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        Long ttl = tryAcquire(leaseTime, unit, Thread.currentThread().getId());
        if (ttl == null) {
            isLocked = true;
            return true;
        }
        try {
            time -= (System.currentTimeMillis() - current);
            if (time <= 0) {
                return false;
            }
            while (true) {
                long currentTime = System.currentTimeMillis();

                ttl = tryAcquire(leaseTime, unit, Thread.currentThread().getId());
                if (ttl == null) {
                    isLocked = true;
                    return true;
                }

                time -= (System.currentTimeMillis() - currentTime);
                if (time <= 0) {
                    return false;
                }
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    semaphore.tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    semaphore.tryAcquire(time, TimeUnit.MILLISECONDS);
                }
                time -= (System.currentTimeMillis() - currentTime);
                if (time <= 0) {
                    return false;
                }
            }
        } catch (InterruptedException e) {
            return false;
        } finally {
            if (semaphore.availablePermits() == 0) {
                semaphore.release();
            }
        }
    }

    @Override
    public void unlock() {
        if (isLocked) {
            Long unlock = unlock(Thread.currentThread().getId());
            if (unlock == null) {
                throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                        + id + " thread-id: " + Thread.currentThread().getId());
            }
        }

    }

    @Override
    public Condition newCondition() {
        return null;
    }

    private Long tryAcquire(long leaseTime, TimeUnit unit, final long threadId) {
        innerLeaseTime = unit.toMillis(leaseTime);
        Jedis jedis = null;
        try {
            List<String> args = new ArrayList<>(2);
            args.add(String.valueOf(innerLeaseTime));
            args.add(getLockField(threadId));

            jedis = jedisPool.getResource();
            return (Long) RedisCommandExecutor.evalWrite(jedis, RedisCommandExecutor.LOCK_SCRIPT, Collections.singletonList(lockName), args);
        } finally {
            if (jedis != null) {
                jedisPool.returnResource(jedis);
            }
        }
    }

    private Long unlock(final long threadId) {
        Jedis jedis = null;
        try {
            List<String> args = new ArrayList<>(2);
            args.add(String.valueOf(innerLeaseTime));
            args.add(getLockField(threadId));

            jedis = jedisPool.getResource();
            return (Long) RedisCommandExecutor.evalWrite(jedis, RedisCommandExecutor.UNLOCK_SCRIPT, Collections.singletonList(lockName), args);
        } finally {
            if (jedis != null) {
                jedisPool.returnResource(jedis);
            }
        }
    }

    public String getLockField(long threadId) {
        return id + ":" + threadId;
    }

    private static final class RedisCommandExecutor {
        public static final String LOCK_SCRIPT = "if (redis.call('exists', KEYS[1]) == 0) then " +
                "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                "return nil; " +
                "end; " +
                "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                "return nil; " +
                "end; " +
                "return redis.call('pttl', KEYS[1]);";

        public static final String UNLOCK_SCRIPT = "if (redis.call('exists', KEYS[1]) == 0) then " +
                "return 1; " +
                "end;" +
                "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                "redis.call('del', KEYS[1]); " +
                "return 1; " +
                "end; " +
                "return nil;";

        public static Object evalWrite(Jedis jedis, String script, List<String> keys, List<String> args) {
            return jedis.eval(script, keys, args);
        }
    }
}
